// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved.
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace WordCount
{
    class AggregateActor : FabricAdHocActorBase, IDisposable
    {
        private int m_numReducers;
        private FabricActorClient m_client;

        public AggregateActor(int numReducers)
            : base("aggregator")
        {
            m_numReducers = numReducers;
        }

        public void InitializeAggregator(FabricActorClient client, string assemblyName, byte[] assemblyBytes)
        {
            m_client = client;
            client.AddAssembly(assemblyName, assemblyBytes);
            client.CallMethod("InitializeAggregator", new object[] { m_numReducers });
        }

        public List<KeyValuePair<string,int>> GetRankedItems(int fromInclusive, int toInclusive)
        {
            var result = m_client.CallMethod<List<KeyValuePair<string,int>>>("GetRankedItems", new object[] {fromInclusive, toInclusive});
            return result;
        }

        public List<Tuple<string, int, int>> GetSearchResults(string searchTerm)
        {
            var result = m_client.CallMethod<List<Tuple<string, int, int>>>("GetSearchResults", new object[] { searchTerm });
            return result;
        }

        public void SubscribeDone(IObserver<ActorEvent> eventObserver)
        {
            m_client.Subscribe("Done", eventObserver);
        }

        public void Dispose()
        {
            if (m_client != null)
            {
                m_client.Dispose();
                m_client = null;
            }
        }



        [ActorMethod]
        public static object InitializeAggregator(dynamic state, dynamic[] args)
        {
            state.numReducers = (int)args[0];
            state.reducersCompleted = 0;
            state.InitializeDictionaryStuff();
            state.Remove("orderedList");
            state.Remove("dictionaryWithStats");
            return null;
        }

        [ActorMethod]
        public static object ProcessAggregatedInfo(dynamic state, dynamic[] args)
        {
            string fname = args[0];
            string justTheName = Path.GetFileNameWithoutExtension(fname);
            BinaryFormatter bf = new BinaryFormatter();
            using (var stream = new FileStream(fname, FileMode.Open, FileAccess.Read))
            {
                var list = (List<KeyValuePair<string, int>>)bf.Deserialize(stream);
                int totalWordCount = 0;
                foreach (var kvp in list)
                {
                    state.SetOrAdd(kvp.Key, kvp.Value);
                    totalWordCount++;
                    if ((totalWordCount % 20) == 0)
                        state.Publish("Update", String.Format("{0}\n{1}", justTheName, totalWordCount));
                }
            }
            return null;
        }

        [ActorMethod]
        public static object ReducingComplete(dynamic state, dynamic[] args)
        {
            var reducersCompleted = (int)state.reducersCompleted;
            var numReducers = (int)state.numReducers;

            reducersCompleted++;
            state.reducersCompleted = reducersCompleted;

            if (reducersCompleted == numReducers)
            {
                var fullList = (List<KeyValuePair<string, int>>)state.EnumerateAll();
                var orderedList = fullList.OrderByDescending(kvp => kvp.Value).ToList();
                state.orderedList = orderedList; // Store a copy
                var dictionaryWithStats = new Dictionary<string, Tuple<int, int>>();
                for (int i = 0; i < orderedList.Count; i++)
                {
                    var kvp = orderedList[i];
                    dictionaryWithStats[kvp.Key] = Tuple.Create(kvp.Value, i);
                }
                state.dictionaryWithStats = dictionaryWithStats; // store it
                state.Publish("Done", "<nothing>");
                state.Publish("Update", "Done");
            }

            return null;
        }

        [ActorMethod]
        public static object GetRankedItems(dynamic state, dynamic[] args)
        {
            if (!state.Contains("orderedList") || state.orderedList == null)
                throw new InvalidOperationException("No ordered list was produced!");

            var orderedList = (List<KeyValuePair<string, int>>)state.orderedList;
            int fromRank = args[0];
            int toRank = args[1];

            if (fromRank < 0) fromRank = 0;
            if (toRank >= orderedList.Count) toRank = orderedList.Count - 1;

            var result = new List<KeyValuePair<string, int>>();
            for (int i = fromRank; i <= toRank; i++) result.Add(orderedList[i]);

            return result;
        }

        [ActorMethod]
        public static object GetSearchResults(dynamic state, dynamic[] args)
        {
            if (!state.Contains("dictionaryWithStats") || state.dictionaryWithStats == null)
                throw new InvalidOperationException("No dictionaryWithStats was produced!");

            var dictionaryWithStats = (Dictionary<string, Tuple<int, int>>)state.dictionaryWithStats;
            //var orderedList = (List<KeyValuePair<string, int>>)state.orderedList;
            string searchTerm = args[0];
            Regex regex = new Regex(searchTerm);

            var result = dictionaryWithStats.Where(kvp => regex.Match(kvp.Key).Success).Select(x => Tuple.Create(x.Key, x.Value.Item1, x.Value.Item2)).ToList();
            return result;
        }

    }
}
